﻿using Confluent.Kafka;
using furion.blog.Application.Contracts;
using furion.blog.Application.KafkaMQ.Producers;
using furion.blog.Domain.Shared.Dto.Sms;
using furion.blog.Domain.Shared.Dto.Wx.MiniProgram;
using Furion;
using Furion.DependencyInjection;
using Microsoft.AspNetCore.Hosting;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace furion.blog.Application.KafkaMQ.Consumers
{
    /// <summary>
    /// 消费微信发送信息给会员
    /// </summary>
    public class WxSendMsgToMemberConsumer :  IDisposable
    {
        string TOPIC = WxSendMsgToMemberProducer.TOPIC;
        string GROUP = WxSendMsgToMemberProducer.TOPIC+"-GROUPONE";

        IConsumer<Ignore, string> _consumer;
        private readonly IWebHostEnvironment _hostingEnvironment;
        private readonly IWxService _wxService;

        public WxSendMsgToMemberConsumer(IWebHostEnvironment hostEnvironment)
        {
            this._hostingEnvironment = hostEnvironment;
            this._wxService = App.GetService<IWxService>();
            //Task t = new Task(StartDeal);
            //t.Start();
            //StartDeal();
            Task.Run(async () => {
                await StartDeal();
            });
        }


        private async Task StartDeal()
        {
            var cts = new CancellationTokenSource();
            Console.CancelKeyPress += (_, e) =>
            {
                e.Cancel = true;
                cts.Cancel();
            };

            var config = new ConsumerConfig
            {
                BootstrapServers = App.Configuration["Kafka:BootstrapServers"],
                GroupId = GROUP,
                //StatisticsIntervalMs = 5000,
                //SessionTimeoutMs = 6000,
                AutoOffsetReset = AutoOffsetReset.Earliest,
                //EnablePartitionEof = true,
                //EnableAutoCommit = true, // (the default)
                EnableAutoOffsetStore = false,

            };

            //const int commitPeriod = 5;

            _consumer = new ConsumerBuilder<Ignore, string>(config)
                                 .SetErrorHandler((_, e) =>
                                 {
                                     Console.WriteLine($"Error: {e.Reason}");
                                 })
                                 .SetStatisticsHandler((_, json) =>
                                 {
                                     Console.WriteLine($" - {DateTime.Now:yyyy-MM-dd HH:mm:ss} > monitoring..");
                                     //Console.WriteLine($"Statistics: {json}");
                                 })
                                 .SetPartitionsAssignedHandler((c, partitions) =>
                                 {
                                     Console.WriteLine($"Assigned partitions: [{string.Join(", ", partitions)}]");
                                 })
                                 .SetPartitionsRevokedHandler((c, partitions) =>
                                 {
                                     Console.WriteLine($"Revoking assignment: [{string.Join(", ", partitions)}]");
                                 })
                                 .Build();
            _consumer.Subscribe(TOPIC);

            try
            {
                while (true)
                {
                    try
                    {
                        var consumeResult = _consumer.Consume(cts.Token);

                        if (consumeResult.Message != null && !string.IsNullOrEmpty(consumeResult.Message.Value))
                        {
                            Console.WriteLine($"Received message at {consumeResult.TopicPartitionOffset}: ${consumeResult.Message.Value}");
                            /*string path = _hostingEnvironment.WebRootPath + @"\logs\test.txt";
                            if (!File.Exists(path)) {
                                File.Create(path).Close();
                            }
                            using (StreamWriter writer = File.AppendText(path))
                            {
                                writer.WriteLine($"Received message at {consumeResult.TopicPartitionOffset}: ${consumeResult.Message.Value}");
                                writer.Flush();//刷新缓存
                                writer.Close();
                                writer.Dispose();
                            }*/
                            //读取内容写入 mongo

                            var obj = JsonConvert.DeserializeObject<SendMsgToMemberDto>(consumeResult.Message.Value);
                            if (obj != null) {
                              await this._wxService.SendMsgToMember(obj);
                            }
                        }
                        _consumer.StoreOffset(consumeResult);//<----this
                    }
                    catch (ConsumeException e)
                    {
                        Console.WriteLine($"Consume error: {e.Error.Reason}");
                    }

                }
            }
            catch (OperationCanceledException)
            {
                Console.WriteLine("Closing consumer.");
                _consumer.Close();
            }

        }


        public void Dispose()
        {
            _consumer?.Dispose();
        }

    }
}